package com.shujia.spark.opt

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

import java.sql.{Connection, DriverManager, PreparedStatement}

object Demo3foreachPartitions {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("cache")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    val sc: SparkContext = spark.sparkContext

    //读取数据
    val linesDS: RDD[String] = sc.textFile("data/students.txt")

    /**
     * 将rdd的数据保存到数据库中
     * 使用foreach 将rdd的数据写入到数据库中
     * 1、在foreach算子内创建网络链接，会导致每一条数据都需要建立链接，效率极低
     * 2、如果将网络链接放在foreach算子的外面，在算子呢内不能使用
     *
     *
     * 在算子内不能使用算子外的一个网络链接
     * 1、网络链接不能在网络中传输（不能序列化）
     *
     */


    /*linesDS.foreach((line: String) => {

      //1、加载驱动
      Class.forName("com.mysql.jdbc.Driver")
      //2、建立链接
      val con: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")
      //3、编写sql
      val stat: PreparedStatement = con.prepareStatement("insert into student(id,name,age,gender,clazz) values(?,?,?,?,?)")


      val split: Array[String] = line.split(",")
      //4、指定字段
      stat.setInt(1, split(0).toInt)
      stat.setString(2, split(1))
      stat.setInt(3, split(2).toInt)
      stat.setString(4, split(3))
      stat.setString(5, split(4))

      //执行插入
      stat.execute()
      stat.close()
      con.close()
    })*/

    /**
     * 使用foreachPartition只会为每一个分区创建一个链接
     */
    linesDS.foreachPartition((partition: Iterator[String]) => {
      println("foreachPartition")

      //每一额分区执行一次
      //1、加载驱动
      Class.forName("com.mysql.jdbc.Driver")
      //2、建立链接
      val con: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")
      //3、编写sql
      val stat: PreparedStatement = con.prepareStatement("insert into student(id,name,age,gender,clazz) values(?,?,?,?,?)")

      partition.foreach((line: String) => {
        val split: Array[String] = line.split(",")
        //4、指定字段
        stat.setInt(1, split(0).toInt)
        stat.setString(2, split(1))
        stat.setInt(3, split(2).toInt)
        stat.setString(4, split(3))
        stat.setString(5, split(4))
        //执行插入
        stat.execute()
      })

      stat.close()
      con.close()
    })


  }

}
